//! HTTP/2 Server Connections

use std::fmt;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use pin_project_lite::pin_project;
use rama_core::extensions::ExtensionsMut;
use rama_core::rt::Executor;
use std::task::ready;
use tokio::io::{AsyncRead, AsyncWrite};

use crate::body::Incoming as IncomingBody;
use crate::proto;
use crate::service::HttpService;

pin_project! {
    /// A [`Future`](core::future::Future) representing an HTTP/2 connection, bound to a
    /// [`Service`](crate::service::Service), returned from
    /// [`Builder::serve_connection`](struct.Builder.html#method.serve_connection).
    ///
    /// To drive HTTP on this connection this future **must be polled**, typically with
    /// `.await`. If it isn't polled, no progress will be made on this connection.
    #[must_use = "futures do nothing unless polled"]
    pub struct Connection<T, S>
    where
        S: HttpService<IncomingBody>,
    {
        conn: proto::h2::Server<T, S>,
    }
}

/// A configuration builder for HTTP/2 server connections.
///
/// **Note**: The default values of options are *not considered stable*. They
/// are subject to change at any time.
#[derive(Clone, Debug)]
pub struct Builder {
    exec: Executor,
    h2_builder: proto::h2::server::Config,
}

// ===== impl Connection =====

impl<I, S> fmt::Debug for Connection<I, S>
where
    S: HttpService<IncomingBody>,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Connection").finish()
    }
}

impl<I, S> Connection<I, S>
where
    S: HttpService<IncomingBody>,
    I: AsyncRead + AsyncWrite + Send + Unpin + ExtensionsMut + 'static,
{
    /// Start a graceful shutdown process for this connection.
    ///
    /// This `Connection` should continue to be polled until shutdown
    /// can finish.
    ///
    /// # Note
    ///
    /// This should only be called while the `Connection` future is still
    /// pending. If called after `Connection::poll` has resolved, this does
    /// nothing.
    pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
        self.conn.graceful_shutdown();
    }
}

impl<I, S> Future for Connection<I, S>
where
    S: HttpService<IncomingBody>,
    I: AsyncRead + AsyncWrite + Send + Unpin + ExtensionsMut + 'static,
{
    type Output = crate::Result<()>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match ready!(Pin::new(&mut self.conn).poll(cx)) {
            Ok(_done) => {
                //TODO: the proto::h2::Server no longer needs to return
                //the Dispatched enum
                Poll::Ready(Ok(()))
            }
            Err(e) => Poll::Ready(Err(e)),
        }
    }
}

// ===== impl Builder =====

impl Builder {
    /// Create a new connection builder.
    #[must_use]
    pub fn new(exec: Executor) -> Self {
        Self {
            exec,
            h2_builder: Default::default(),
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Configures the maximum number of pending reset streams allowed before a GOAWAY will be sent.
        ///
        /// This will default to the default value set by the [`h2` crate](https://crates.io/crates/h2).
        /// As of v0.4.0, it is 20.
        ///
        /// See <https://github.com/hyperium/hyper/issues/2877> for more information.
        pub fn max_pending_accept_reset_streams(mut self, max: Option<usize>) -> Self {
            self.h2_builder.max_pending_accept_reset_streams = max;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Configures the maximum number of local reset streams allowed before a GOAWAY will be sent.
        ///
        /// If not set, rama_http_core will use a default, currently of 1024.
        ///
        /// If `None` is supplied, rama_http_core will not apply any limit.
        /// This is not advised, as it can potentially expose servers to DOS vulnerabilities.
        ///
        /// See <https://rustsec.org/advisories/RUSTSEC-2024-0003.html> for more information.
        pub fn max_local_error_reset_streams(mut self, max: Option<usize>) -> Self {
            self.h2_builder.max_local_error_reset_streams = max;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
        /// stream-level flow control.
        ///
        /// If not set, rama_http_core will use a default.
        ///
        /// [spec]: https://httpwg.org/specs/rfc9113.html#SETTINGS_INITIAL_WINDOW_SIZE
        pub fn initial_stream_window_size(mut self, sz: u32) -> Self {
            self.h2_builder.adaptive_window = false;
            self.h2_builder.initial_stream_window_size = sz;

            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets the max connection-level flow control for HTTP2.
        ///
        /// If not set, rama_http_core will use a default.
        pub fn initial_connection_window_size(mut self, sz: u32) -> Self {
            self.h2_builder.adaptive_window = false;
            self.h2_builder.initial_conn_window_size = sz;

            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets whether to use an adaptive flow control.
        ///
        /// Enabling this will override the limits set in
        /// `initial_stream_window_size` and
        /// `initial_connection_window_size`.
        pub fn adaptive_window(mut self, enabled: bool) -> Self {
            self.h2_builder.adaptive_window = enabled;
            if enabled {
                self.h2_builder.initial_conn_window_size = proto::h2::SPEC_WINDOW_SIZE;
                self.h2_builder.initial_stream_window_size = proto::h2::SPEC_WINDOW_SIZE;
            }
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets the maximum frame size to use for HTTP2.
        ///
        /// If not set, rama_http_core will use a default.
        pub fn max_frame_size(mut self, sz: u32) -> Self {
            self.h2_builder.max_frame_size = sz;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets the [`SETTINGS_MAX_CONCURRENT_STREAMS`][spec] option for HTTP2
        /// connections.
        ///
        /// Default is 200, but not part of the stability of rama_http_core. It could change
        /// in a future release. You are encouraged to set your own limit.
        ///
        /// Passing `None` will remove any limit.
        ///
        /// [spec]: https://httpwg.org/specs/rfc9113.html#SETTINGS_MAX_CONCURRENT_STREAMS
        pub fn max_concurrent_streams(mut self, max: Option<u32>) -> Self {
            self.h2_builder.max_concurrent_streams = max;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets an interval for HTTP2 Ping frames should be sent to keep a
        /// connection alive.
        ///
        /// Pass `None` to disable HTTP2 keep-alive.
        ///
        /// Default is currently disabled.
        pub fn keep_alive_interval(mut self, interval: Option<Duration>) -> Self {
            self.h2_builder.keep_alive_interval = interval;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
        ///
        /// If the ping is not acknowledged within the timeout, the connection will
        /// be closed. Does nothing if `keep_alive_interval` is disabled.
        ///
        /// Default is 20 seconds.
        pub fn keep_alive_timeout(mut self, timeout: Duration) -> Self {
            self.h2_builder.keep_alive_timeout = timeout;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Set the maximum write buffer size for each HTTP/2 stream.
        ///
        /// Default is currently ~400KB, but may change.
        pub fn max_send_buf_size(mut self, max: u32) -> Self {
            self.h2_builder.max_send_buffer_size = max;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Enables the [extended CONNECT protocol].
        ///
        /// [extended CONNECT protocol]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
        pub fn enable_connect_protocol(mut self) -> Self {
            self.h2_builder.enable_connect_protocol = true;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Sets the max size of received header frames.
        ///
        /// Default is currently 16KB, but can change.
        pub fn max_header_list_size(mut self, max: u32) -> Self {
            self.h2_builder.max_header_list_size = max;
            self
        }
    }

    rama_utils::macros::generate_set_and_with! {
        /// Set whether the `date` header should be included in HTTP responses.
        ///
        /// Note that including the `date` header is recommended by RFC 7231.
        ///
        /// Default is `true`.
        pub fn auto_date_header(mut self, enabled: bool) -> Self {
            self.h2_builder.date_header = enabled;
            self
        }
    }

    /// Bind a connection together with a [`Service`](crate::service::Service).
    ///
    /// This returns a Future that must be polled in order for HTTP to be
    /// driven on the connection.
    pub fn serve_connection<S, I>(&self, io: I, service: S) -> Connection<I, S>
    where
        S: HttpService<IncomingBody>,
        I: AsyncRead + AsyncWrite + Send + Unpin + ExtensionsMut + 'static,
    {
        let proto = proto::h2::Server::new(io, service, &self.h2_builder, self.exec.clone());
        Connection { conn: proto }
    }
}

#[cfg(test)]
mod tests {
    use crate::service::VoidHttpService;
    use tokio::net::TcpStream;

    use super::*;

    #[test]
    fn test_assert_send_static() {
        fn g<T: Send + 'static>() {}
        g::<Connection<TcpStream, VoidHttpService>>();
    }
}
